package com.bst.etl.datax.service.impl;

import com.alibaba.druid.pool.DruidDataSource;
import com.bst.common.constant.JobConstant;
import com.bst.datatx.core.biz.model.JobParam;
import com.bst.datatx.core.biz.model.ReturnT;
import com.bst.datatx.core.enums.ExecutorBlockStrategyEnum;
import com.bst.datatx.core.glue.GlueTypeEnum;
import com.bst.datatx.core.util.Constants;
import com.bst.etl.datax.core.conf.JobAdminConfig;
import com.bst.etl.datax.core.cron.CronExpression;
import com.bst.etl.datax.core.route.ExecutorRouteStrategyEnum;
import com.bst.etl.datax.core.thread.JobTriggerPoolHelper;
import com.bst.etl.datax.core.trigger.TriggerTypeEnum;
import com.bst.etl.datax.core.util.I18nUtil;
import com.bst.etl.datax.dto.TriggerJobDto;
import com.bst.etl.datax.entity.JobGroup;
import com.bst.etl.datax.entity.JobLog;
import com.bst.etl.datax.entity.JobTemplate;
import com.bst.etl.datax.mapper.JobGroupMapper;
import com.bst.etl.datax.mapper.JobLogGlueMapper;
import com.bst.etl.datax.mapper.JobLogMapper;
import com.bst.etl.datax.mapper.JobTemplateMapper;
import com.bst.etl.datax.service.JobTemplateService;
import com.bst.etl.datax.util.EtlUtil;
import com.bst.etl.domain.EtlDrawBaseDatax;
import com.bst.etl.domain.EtlDrawDm;
import com.bst.etl.domain.EtlDrawDw;
import com.bst.framework.datasource.DynamicDataSource;
import org.springframework.stereotype.Service;

import jakarta.annotation.Resource;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * core job action for xxl-job
 *
 * @author xuxueli 2016-5-28 15:30:33
 */
@Service("jobTemplateService")
public class JobTemplateServiceImpl implements JobTemplateService {
    @Resource
    private JobGroupMapper jobGroupMapper;
    @Resource
    private JobTemplateMapper jobTemplateMapper;
    @Resource
    private JobLogMapper jobLogMapper;
    @Resource
    private JobLogGlueMapper jobLogGlueMapper;

    @Override
    public List<JobTemplate> selectJobTemplateList(JobTemplate jobTemplate) {
        // page list
        List<JobTemplate> list = jobTemplateMapper.selectJobTemplateList(jobTemplate);
        return list;
    }

    @Override
    public ReturnT<String> add(JobTemplate jobTemplate) {
        // valid
        JobGroup group = jobGroupMapper.load(jobTemplate.getJobGroup());
        if (group == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("system_please_choose") + I18nUtil.getString("jobinfo_field_jobgroup")));
        }
        if (!CronExpression.isValidExpression(jobTemplate.getJobCron())) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobinfo_field_cron_invalid"));
        }
        if (jobTemplate.getJobDesc() == null || jobTemplate.getJobDesc().trim().length() == 0) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("system_please_input") + I18nUtil.getString("jobinfo_field_jobdesc")));
        }
        if (jobTemplate.getUserId() == 0) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("system_please_input") + I18nUtil.getString("jobinfo_field_author")));
        }
        if (ExecutorRouteStrategyEnum.match(jobTemplate.getExecutorRouteStrategy(), null) == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_executorRouteStrategy") + I18nUtil.getString("system_invalid")));
        }
        if (ExecutorBlockStrategyEnum.match(jobTemplate.getExecutorBlockStrategy(), null) == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_executorBlockStrategy") + I18nUtil.getString("system_invalid")));
        }
        if (GlueTypeEnum.match(jobTemplate.getGlueType()) == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_gluetype") + I18nUtil.getString("system_invalid")));
        }
        if (GlueTypeEnum.BEAN == GlueTypeEnum.match(jobTemplate.getGlueType()) && (jobTemplate.getExecutorHandler() == null || jobTemplate.getExecutorHandler().trim().length() == 0)) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("system_please_input") + "JobHandler"));
        }

        // fix "\r" in shell
        if (GlueTypeEnum.GLUE_SHELL == GlueTypeEnum.match(jobTemplate.getGlueType()) && jobTemplate.getGlueSource() != null) {
            jobTemplate.setGlueSource(jobTemplate.getGlueSource().replaceAll("\r", ""));
        }

        // add in db
        jobTemplate.setAddTime(new Date());
        jobTemplate.setUpdateTime(new Date());
        jobTemplate.setGlueUpdatetime(new Date());
        jobTemplateMapper.save(jobTemplate);
        if (jobTemplate.getId() < 1) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_add") + I18nUtil.getString("system_fail")));
        }

        return new ReturnT<String>(String.valueOf(jobTemplate.getId()));
    }

    private boolean isNumeric(String str) {
        try {
            int result = Integer.valueOf(str);
            return true;
        } catch (NumberFormatException e) {
            return false;
        }
    }

    @Override
    public ReturnT<String> update(JobTemplate jobTemplate) {

        // valid
        if (!CronExpression.isValidExpression(jobTemplate.getJobCron())) {
            return new ReturnT<>(ReturnT.FAIL_CODE, I18nUtil.getString("jobinfo_field_cron_invalid"));
        }
        if (jobTemplate.getJobDesc() == null || jobTemplate.getJobDesc().trim().length() == 0) {
            return new ReturnT<>(ReturnT.FAIL_CODE, (I18nUtil.getString("system_please_input") + I18nUtil.getString("jobinfo_field_jobdesc")));
        }
        if (jobTemplate.getUserId() == 0) {
            return new ReturnT<>(ReturnT.FAIL_CODE, (I18nUtil.getString("system_please_input") + I18nUtil.getString("jobinfo_field_author")));
        }
        if (ExecutorRouteStrategyEnum.match(jobTemplate.getExecutorRouteStrategy(), null) == null) {
            return new ReturnT<>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_executorRouteStrategy") + I18nUtil.getString("system_invalid")));
        }
        if (ExecutorBlockStrategyEnum.match(jobTemplate.getExecutorBlockStrategy(), null) == null) {
            return new ReturnT<>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_executorBlockStrategy") + I18nUtil.getString("system_invalid")));
        }

        // ChildJobId valid
        if (jobTemplate.getChildJobId() != null && jobTemplate.getChildJobId().trim().length() > 0) {
            String[] childJobIds = jobTemplate.getChildJobId().split(",");
            for (String childJobIdItem : childJobIds) {
                if (childJobIdItem != null && childJobIdItem.trim().length() > 0 && isNumeric(childJobIdItem)) {
                    JobTemplate childJobTemplate = jobTemplateMapper.loadById(Integer.parseInt(childJobIdItem));
                    if (childJobTemplate == null) {
                        return new ReturnT<String>(ReturnT.FAIL_CODE,
                                MessageFormat.format((I18nUtil.getString("jobinfo_field_childJobId") + "({0})" + I18nUtil.getString("system_not_found")), childJobIdItem));
                    }
                } else {
                    return new ReturnT<String>(ReturnT.FAIL_CODE,
                            MessageFormat.format((I18nUtil.getString("jobinfo_field_childJobId") + "({0})" + I18nUtil.getString("system_invalid")), childJobIdItem));
                }
            }

            // join , avoid "xxx,,"
            String temp = Constants.STRING_BLANK;
            for (String item : childJobIds) {
                temp += item + Constants.SPLIT_COMMA;
            }
            temp = temp.substring(0, temp.length() - 1);

            jobTemplate.setChildJobId(temp);
        }

        // group valid
        JobGroup jobGroup = jobGroupMapper.load(jobTemplate.getJobGroup());
        if (jobGroup == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_jobgroup") + I18nUtil.getString("system_invalid")));
        }

        // stage job info
        JobTemplate exists_jobTemplate = jobTemplateMapper.loadById(jobTemplate.getId());
        if (exists_jobTemplate == null) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("jobinfo_field_id") + I18nUtil.getString("system_not_found")));
        }

        // next trigger time (5s后生效，避开预读周期)
        long nextTriggerTime = exists_jobTemplate.getTriggerNextTime();

        exists_jobTemplate.setJobGroup(jobTemplate.getJobGroup());
        exists_jobTemplate.setJobCron(jobTemplate.getJobCron());
        exists_jobTemplate.setJobDesc(jobTemplate.getJobDesc());
        exists_jobTemplate.setUserId(jobTemplate.getUserId());
        exists_jobTemplate.setAlarmEmail(jobTemplate.getAlarmEmail());
        exists_jobTemplate.setExecutorRouteStrategy(jobTemplate.getExecutorRouteStrategy());
        exists_jobTemplate.setExecutorHandler(jobTemplate.getExecutorHandler());
        exists_jobTemplate.setExecutorParam(jobTemplate.getExecutorParam());
        exists_jobTemplate.setExecutorBlockStrategy(jobTemplate.getExecutorBlockStrategy());
        exists_jobTemplate.setExecutorTimeout(jobTemplate.getExecutorTimeout());
        exists_jobTemplate.setExecutorFailRetryCount(jobTemplate.getExecutorFailRetryCount());
        exists_jobTemplate.setChildJobId(jobTemplate.getChildJobId());
        exists_jobTemplate.setTriggerNextTime(nextTriggerTime);
        exists_jobTemplate.setJobJson(jobTemplate.getJobJson());
        exists_jobTemplate.setJvmParam(jobTemplate.getJvmParam());
        exists_jobTemplate.setProjectId(jobTemplate.getProjectId());
        exists_jobTemplate.setUpdateTime(new Date());
        exists_jobTemplate.setTriggerStatus(jobTemplate.getTriggerStatus());
        jobTemplateMapper.update(exists_jobTemplate);

        return ReturnT.SUCCESS;
    }

    @Override
    public ReturnT<String> remove(int id) {
        JobTemplate xxlJobTemplate = jobTemplateMapper.loadById(id);
        if (xxlJobTemplate == null) {
            return ReturnT.SUCCESS;
        }

        jobTemplateMapper.delete(id);
        jobLogGlueMapper.deleteByJobId(id);
        return ReturnT.SUCCESS;
    }

    @Override
    public void jobTemplateTrigger(TriggerJobDto dto,TriggerTypeEnum triggerType) {
        String executorParam=dto.getExecutorParam();

        if (executorParam == null) {
            executorParam = "";
        }
        if(TriggerTypeEnum.CRON.equals(triggerType)) {
            dto.setType("all");
        }
        if("all".equals(dto.getType())) {
            JobLog logModel = new JobLog();
            logModel.setJobGroup(dto.getTemplateId());
            logModel.setTriggerTime(new Date());
            dto.setType(JobConstant.JOB_TASK_BASE_DATAX);
            if(queryData(dto).size()>0) {
                JobTriggerPoolHelper.trigger(dto,triggerType);
                waitTaskEnd(logModel,dto);
            }
            dto.setType(JobConstant.JOB_TASK_BASE_DW);
            if(queryData(dto).size()>0) {
                JobTriggerPoolHelper.trigger(dto,triggerType);
                waitTaskEnd(logModel,dto);
            }
            dto.setType(JobConstant.JOB_TASK_BASE_DM);
            if(queryData(dto).size()>0) {
                JobTriggerPoolHelper.trigger(dto,triggerType);
                waitTaskEnd(logModel,dto);
            }
        } else {
            if(queryData(dto).size()>0) {
                JobTriggerPoolHelper.trigger(dto,triggerType);
            }
        }
    }

    private List<JobParam> queryData(TriggerJobDto dto) {
        List<EtlDrawBaseDatax> dataxList;
        List<EtlDrawDw> dwList;
        List<EtlDrawDm> dmList;
        List<JobParam> jobParams = new ArrayList<>();
        Integer templateId = dto.getTemplateId();
        if("datax".equals(dto.getType())) {
            dataxList = JobAdminConfig.getAdminConfig().getEtlDrawBaseDataxMapper().selectEtlDrawBaseDataxByIdTemplate(templateId,dto.getNaFilter());
            if(dataxList.size()>0) {
                jobParams.addAll(dataxToJobParam(dataxList,dto));
            }
        } else if("dw".equals(dto.getType())) {
            dwList = JobAdminConfig.getAdminConfig().getEtlDrawDwMapper().selectEtlDrawDwByIdTemplate(templateId);
            if(dwList.size()>0) {
                jobParams.addAll(dwToJobParam(dwList,dto));
            }
       }//  else if("dm".equals(dto.getType())) {
//            dmList = JobAdminConfig.getAdminConfig().getEtlDrawDmMapper().selectEtlDrawDmByIdTemplate(templateId);
//            if(dmList.size()>0) {
//                jobParams.addAll(dmToJobParam(dmList,dto));
//            }
//        }
        dto.setJobParams(jobParams);
        return jobParams;
    }

    private void waitTaskEnd(JobLog logModel,TriggerJobDto dto) {

        int times = 1*60*1000;
        int num=0;
        try {
            List<JobLog> logList = new ArrayList<>();
            while(logList.size()==0 && num<20) {
                num = num+1;
                Thread.sleep(10*1000); //休眠10秒等待定时数据生成
                logList = jobLogMapper.selectNewJobLog(logModel);
                if(logList.size()==0) {
                    System.out.println(num*10+"秒数据未生成");
                }
            }
            logList = jobLogMapper.selectNotEndJobLog(logModel);
            while (logList.size()>0 && num<60 ) {
                num = num+1;
                Thread.sleep(times);
                System.out.println("休眠第"+(num-1)+"次,剩余"+logList.size()+"条记录");
                logList = jobLogMapper.selectNotEndJobLog(logModel);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("任务结束");
        }
    }


    private static List<JobParam> dataxToJobParam(List<EtlDrawBaseDatax> dataxes,TriggerJobDto dto) {
        List<JobParam> params=new ArrayList<>();JobParam param;
        List<Long> jobIds = dto.getJobIds();
        for (EtlDrawBaseDatax datax : dataxes) {
            if(jobIds!=null && jobIds.size()>0) {
                if(!jobIds.contains(datax.getIdDrawDatax())) {
                    continue;
                }
            }
            param = new JobParam();
            param.setJobId(datax.getIdDrawDatax());
            param.setJobName(datax.getNa());
            param.setTableName(datax.getTbTar());
            param.setDayBegin(dto.getDayBegin());
            param.setDayEnd(dto.getDayEnd());
            param.setJsonConfig(datax.getConfig());
            param.setEuType("datax");
            String json = datax.getConfig();
            json = EtlUtil.dataXConfigVariHandler(dto.getDayBegin(),dto.getDayEnd(),json);
            param.setJsonConfig(json);
            params.add(param);
        }
        return params;
    }

    private static List<JobParam> dwToJobParam(List<EtlDrawDw> dws,TriggerJobDto dto) {
        List<JobParam> params=new ArrayList<>();JobParam param;
        List<Long> jobIds = dto.getJobIds();
        for (EtlDrawDw dw : dws) {
            if(jobIds!=null && jobIds.size()>0) {
                if(!jobIds.contains(dw.getIdDw())) {
                    continue;
                }
            }
            param = new JobParam();
            param.setDelSql(dw.getSqlDel());
            param.setJobId(dw.getIdDw());
            param.setJobName(dw.getNa());
            param.setDayBegin(dto.getDayBegin());
            param.setDayEnd(dto.getDayEnd());
            param.setInsertSql(dw.getSqlInsert());
            param.setEuSqlExecType(dw.getEuExecType());
            param.setQuerySql(dw.getSqlQry());
            param.setEuType("db");
            DruidDataSource source = DynamicDataSource.getDataSourceById("dw");
            param.setUsername(source.getUsername());
            param.setPassword(source.getPassword());
            param.setUrl(source.getUrl());
            param.setTableName(dw.getCdTb());
            params.add(param);
        }
        return params;
    }

//    private static List<JobParam> dmToJobParam(List<EtlDrawDm> dms, TriggerJobDto dto) {
//        List<JobParam> params=new ArrayList<>();JobParam param;
//        List<Long> jobIds = dto.getJobIds();
//        for (EtlDrawDm dm : dms) {
//            if(jobIds!=null && jobIds.size()>0) {
//                if(!jobIds.contains(dm.getIdDrawDm())) {
//                    continue;
//                }
//            }
//            param = new JobParam();
//            param.setJobId(dm.getIdDrawDm());
//            param.setJobName(dm.getNaIndex());
//            param.setDayBegin(dto.getDayBegin());
//            param.setDayEnd(dto.getDayEnd());
//            param.setInsertSql(dm.getSqlInsert());
//            param.setEuSqlExecType("date");
//            param.setEuType("db");
//            DruidDataSource source = DynamicDataSource.getDataSourceById("dw");
//            param.setUsername(source.getUsername());
//            param.setPassword(source.getPassword());
//            param.setUrl(source.getUrl());
//            param.setTableName(dm.getCdTb());
//            params.add(param);
//        }
//        return params;
//    }
}
